import asyncio
import websockets
import json
import pyaudio
import queue
import numpy as np
import sys
import os
import ctypes
import random

# 尝试加载opus库进行音频解码和编码
try:
    # 强制设置环境变量来帮助opuslib找到正确的DLL
    if sys.platform == 'win32':
        # 获取opus.dll的绝对路径
        opus_path = os.path.join(os.path.dirname(__file__), "xxx xiaozhi-web-client-main", "libs", "windows")
        print(f"opus.dll目录: {opus_path}")
        if os.path.exists(opus_path):
            # 将opus.dll所在目录添加到PATH环境变量
            os.environ['PATH'] = opus_path + ";" + os.environ['PATH']
            print(f"已将opus.dll目录添加到PATH环境变量")
        
        # 直接设置opuslib将使用的DLL路径
        opus_dll_path = os.path.join(opus_path, "opus.dll")
        if os.path.exists(opus_dll_path):
            print(f"确认opus.dll文件存在: {opus_dll_path}")
            # 尝试预加载opus.dll
            try:
                opus_lib = ctypes.cdll.LoadLibrary(opus_dll_path)
                print(f"[OK] 成功预加载opus.dll")
            except Exception as e:
                print(f"[ERROR] 预加载opus.dll失败: {e}")
        else:
            print(f"❌ opus.dll文件不存在: {opus_dll_path}")
    
    # 尝试导入opuslib
    try:
        import opuslib
        print("[OK] 成功导入opuslib")
        OPUS_AVAILABLE = True
        
        # 输出opuslib版本信息
        print(f"opuslib版本: {getattr(opuslib, '__version__', 'unknown')}")
    except ImportError as e:
        print(f"[ERROR] 导入opuslib失败: {e}")
        print("尝试自动安装opuslib...")
        # 尝试安装opuslib
        try:
            import subprocess
            subprocess.check_call([sys.executable, "-m", "pip", "install", "opuslib==3.0.1"])
            import opuslib
            print("[OK] 成功安装并导入opuslib")
            OPUS_AVAILABLE = True
        except Exception as install_e:
            print(f"[ERROR] 安装opuslib失败: {install_e}")
            print("请手动安装opuslib: pip install opuslib==3.0.1")
            OPUS_AVAILABLE = False
except Exception as e:
    print(f"[ERROR] 设置音频库时出错: {e}")
    import traceback
    traceback.print_exc()
    OPUS_AVAILABLE = False

# 音频队列和配置
AUDIO_QUEUE = asyncio.Queue(maxsize=50)  # 使用异步队列以提高性能
RECORDING_QUEUE = queue.Queue(maxsize=50)  # 增加录音队列大小到500，避免数据丢弃
# 默认音频配置
DEFAULT_FORMAT = pyaudio.paInt16  # 16位整数格式
DEFAULT_CHANNELS = 1  # 单声道
DEFAULT_RATE = 16000  # 默认采样率
CHUNK_SIZE = 960  # 缓冲区大小：960样本（约60ms）
MAX_QUEUE_SIZE = 500  # 最大队列大小，与录音队列匹配
# 音频会话状态
is_first_audio_packet = True  # 标记是否为第一个音频数据包
is_recording = False  # 录音状态标志

# 当前使用的音频参数
current_format = DEFAULT_FORMAT
current_channels = DEFAULT_CHANNELS
current_rate = DEFAULT_RATE

# 音频流状态
stream_active = False

# 创建Opus解码器
try:
    if OPUS_AVAILABLE:
        try:
            opus_decoder = opuslib.Decoder(DEFAULT_RATE, DEFAULT_CHANNELS)
            print("[OK] Opus解码器已初始化")
        except Exception as e:
            print(f"[ERROR] 创建Opus解码器失败: {e}")
            opus_decoder = None
    else:
        opus_decoder = None
except Exception as e:
    print(f"[ERROR] 创建Opus解码器时出错: {e}")
    opus_decoder = None

# 录音任务：从麦克风获取音频数据并放入队列，使用与Opus兼容的格式
async def record_audio():
    """录音函数，从麦克风获取音频数据并放入队列，使用与Opus编码器兼容的格式"""
    # 首先声明全局变量
    global is_recording
    # 设置为True
    is_recording = True
    stream = None
    p = None
    
    try:
        # 使用独立的PyAudio实例
        p = pyaudio.PyAudio()
        print("[OK] 录音 - PyAudio已初始化")
        
        # 列出可用的输入设备并选择合适的设备
        device_index = None
        for i in range(p.get_device_count()):
            dev_info = p.get_device_info_by_index(i)
            if dev_info['maxInputChannels'] > 0:
                print(f"[AUDIO] 可用输入设备 {i}: {dev_info['name']}")
                if device_index is None:
                    device_index = i
        
        print(f"[AUDIO] 使用输入设备索引: {device_index}")
        
        # 尝试打开音频输入流 - 使用与Opus兼容的参数：960采样点/帧 (16kHz * 60ms)
        try:
            # 确保使用960采样点，这与Opus编码器完全兼容
            stream = p.open(
                format=DEFAULT_FORMAT,  # 16位整数格式
                channels=DEFAULT_CHANNELS,  # 单声道
                rate=DEFAULT_RATE,  # 16kHz采样率
                input=True,
                frames_per_buffer=960,  # 明确设置为960，确保与Opus兼容
                input_device_index=device_index,  # 使用检测到的设备
                start=False  # 延迟启动
            )
            print(f"[OK] 录音流已创建 - {DEFAULT_RATE}Hz, {DEFAULT_CHANNELS}声道, 960采样点/帧 (Opus兼容)")
        except Exception as stream_e:
            print(f"[ERROR] 创建录音流失败: {stream_e}")
            # 如果无法创建录音流，保持任务运行但不进行录音
            stream = None
        
        record_count = 0
        stream_started = False
        
        while True:
            # 检查录音状态并相应地控制流
            if is_recording and stream is not None and not stream_started:
                try:
                    stream.start_stream()
                    stream_started = True
                    print("[AUDIO] 录音流已启动")
                except Exception as start_e:
                    print(f"[ERROR] 启动录音流失败: {start_e}")
                    stream_started = False
                    await asyncio.sleep(0.1)
                    continue
            elif not is_recording and stream_started:
                try:
                    stream.stop_stream()
                    stream_started = False
                    print("[AUDIO] 录音流已停止")
                except Exception as stop_e:
                    print(f"[ERROR] 停止录音流失败: {stop_e}")
            
            # 只有在录音状态且流已启动时才读取数据
            if is_recording and stream_started:
                try:
                    # 使用非阻塞方式读取，避免阻塞事件循环
                    audio_data = await asyncio.to_thread(stream.read, 960, exception_on_overflow=False)
                    record_count += 1
                    
                    # 验证数据长度，确保每个块都是正确的大小
                    expected_size = 960 * 2  # 960采样点，每个采样点2字节（16位）
                    if len(audio_data) != expected_size:
                        print(f"[WARN] 录音数据长度不匹配: {len(audio_data)} vs 预期 {expected_size}")
                    
                    # 将录音数据放入队列
                    try:
                        # 使用阻塞放入，但设置较短的超时时间，避免长时间阻塞
                        RECORDING_QUEUE.put(audio_data, block=True, timeout=0.1)
                        # 降低日志频率以减少输出
                        if record_count % 20 == 0:
                            print(f"[AUDIO] 已录制 {record_count} 个音频块 (Opus兼容格式)")
                            # 显示队列使用情况
                            print(f"[AUDIO] 录音队列状态: {RECORDING_QUEUE.qsize()}/{RECORDING_QUEUE.maxsize} 个块")
                    except queue.Full:
                        # 队列满时，尝试等待但不丢弃数据
                        print(f"[WARN] 录音队列已满({RECORDING_QUEUE.qsize()}/{RECORDING_QUEUE.maxsize})，正在等待空间...")
                        # 尝试再次放入，使用更长一点的超时时间
                        try:
                            RECORDING_QUEUE.put(audio_data, block=True, timeout=0.5)
                            print("[AUDIO] 队列空间已获得，继续录制")
                        except queue.Full:
                            # 作为最后的手段，尝试清空一小部分旧数据来腾出空间
                            print(f"[CRITICAL] 队列持续满，尝试清理旧数据")
                            # 只清理一个旧数据块，保留大部分录音内容
                            if not RECORDING_QUEUE.empty():
                                try:
                                    RECORDING_QUEUE.get_nowait()
                                    RECORDING_QUEUE.task_done()
                                    # 现在尝试放入新数据
                                    RECORDING_QUEUE.put_nowait(audio_data)
                                    print("[AUDIO] 已清理一个旧数据块，新数据已保存")
                                except:
                                    print(f"[ERROR] 无法保存录音数据，数据丢失")
                except Exception as read_e:
                    print(f"[ERROR] 读取音频数据时出错: {read_e}")
                    # 短暂休眠后继续
                    await asyncio.sleep(0.01)
            
            # 短暂休眠以避免CPU占用过高
            await asyncio.sleep(0.01)
            
    except Exception as e:
        # 录音任务的错误不应影响主程序运行
        print(f"[ERROR] 录音任务出错: {str(e)[:100]}...")
        # 确保is_recording标志为False
        is_recording = False
    finally:
        # 确保资源正确释放
        print("[AUDIO] 正在关闭录音设备...")
        is_recording = False
        if stream is not None:
            try:
                if stream.is_active():
                    stream.stop_stream()
                stream.close()
            except Exception as stream_e:
                print(f"[ERROR] 关闭音频流时出错: {stream_e}")
        if p is not None:
            try:
                p.terminate()
            except Exception as p_e:
                print(f"[ERROR] 终止PyAudio时出错: {p_e}")
        print("[AUDIO] 录音设备已关闭")
        
        # 清空录音队列
        try:
            cleared_count = 0
            while not RECORDING_QUEUE.empty():
                RECORDING_QUEUE.get_nowait()
                RECORDING_QUEUE.task_done()
                cleared_count += 1
            if cleared_count > 0:
                print(f"[AUDIO] 已清空录音队列中的 {cleared_count} 个数据块")
        except Exception as queue_e:
            print(f"[ERROR] 清空录音队列时出错: {queue_e}")

# 处理Opus音频数据
def process_opus_audio(opus_data):
    """将Opus编码的音频数据解码为PCM，即使Opus不可用也返回可播放的音频数据"""
    if not OPUS_AVAILABLE or opus_decoder is None:
        # 如果Opus不可用，返回原始数据作为后备方案
        print("[WARN] Opus解码不可用，使用原始数据作为后备方案")
        # 创建一个简单的音频信号（噪声）作为后备方案
        # 这样即使无法解码，也能听到一些声音表示收到了音频数据
        import random
        fallback_data = bytearray()
        for _ in range(min(len(opus_data) * 2, CHUNK_SIZE)):  # 粗略估计样本数，不超过标准块大小
            # 生成一个小幅度的随机值 (-200到200)，创建可听见的提示音
            val = random.randint(-200, 200)
            # 转换为16位小端格式
            fallback_data.extend(val.to_bytes(2, 'little', signed=True))
        return bytes(fallback_data)
    
    try:
        # 解码Opus数据
        # 使用960样本（60ms）作为解码帧大小
        pcm_data = opus_decoder.decode(opus_data, 960)
        # print(f"[OK] 成功解码音频数据，输出大小: {len(pcm_data)}字节")
        return pcm_data
    except Exception as e:
        print(f"[ERROR] Opus解码错误: {e}")
        # 如果解码失败，同样返回一个简单的音频信号
        import random
        fallback_data = bytearray(CHUNK_SIZE * 2)  # 一个标准块的长度
        for i in range(CHUNK_SIZE):
            val = random.randint(-100, 100)
            fallback_data[i*2:i*2+2] = val.to_bytes(2, 'little', signed=True)
        return bytes(fallback_data)

# 音频播放任务：从队列中获取音频数据并播放
async def play_audio():
    global stream_active, current_format, current_channels, current_rate
    stream_active = True
    stream = None
    p = None
    
    try:
        # 初始化PyAudio
        p = pyaudio.PyAudio()
        print("[OK] PyAudio已初始化")
        
        # 列出所有可用的音频设备
        print("可用音频输出设备:")
        for i in range(p.get_device_count()):
            dev_info = p.get_device_info_by_index(i)
            if dev_info['maxOutputChannels'] > 0:
                print(f"  设备 {i}: {dev_info['name']}")
        
        # 尝试打开音频输出流，优化延迟参数
        stream = p.open(
            format=current_format,
            channels=current_channels,
            rate=current_rate,
            output=True,
            frames_per_buffer=CHUNK_SIZE,
            output_device_index=None,  # 使用默认设备
            start=True  # 直接启动流，减少延迟
        )
        print(f"[OK] 音频流已创建 - {current_rate}Hz, {current_channels}声道, {CHUNK_SIZE}缓冲区")
        
        # 预加载并预热音频设备
        print("[AUDIO] 预热音频设备...")
        
        # 简化的音频系统预热
        # 创建一个激活音频系统的前导信号（小波形而非纯静音）
        lead_in = bytearray(32)
        for i in range(16):
            lead_in[i*2:i*2+2] = (5 if i % 2 == 0 else -5).to_bytes(2, 'little', signed=True)
        
        # 播放激活信号
        await asyncio.to_thread(stream.write, bytes(lead_in))
        await asyncio.sleep(0.01)
        
        # 播放一个标准大小的静音块完成初始化
        full_block = b'\x00' * (CHUNK_SIZE * current_channels * 2)
        await asyncio.to_thread(stream.write, full_block)
        
        print("[OK] 音频设备预热完成")
        
        play_count = 0  # 跟踪播放的音频块数量
        consecutive_empty_data = 0  # 连续空数据计数
        global is_first_audio_packet
        is_first_audio_packet = True  # 重置第一个包标记
        
        while True:
            try:
                # 从队列获取音频数据，设置超时避免死锁
                audio_data = await asyncio.wait_for(
                    AUDIO_QUEUE.get(), timeout=1.0)
                
                # 重置空数据计数
                consecutive_empty_data = 0
                
                play_count += 1
                
                if audio_data is None:  # 结束信号
                    print("[DONE] 收到音频播放结束信号")
                    break
                
                # 检查数据是否为空
                if not audio_data or len(audio_data) == 0:
                    print(f"[WARN] 收到空音频数据块 #{play_count}")
                    consecutive_empty_data += 1
                    # 生成一个简单的提示音
                    empty_audio = bytearray(CHUNK_SIZE * 2)
                    for i in range(CHUNK_SIZE):
                        # 生成一个小的方波信号作为提示
                        val = 100 if i % 10 < 5 else -100
                        empty_audio[i*2:i*2+2] = val.to_bytes(2, 'little', signed=True)
                    audio_data = bytes(empty_audio)
                
                # 确保数据长度适合缓冲区
                if len(audio_data) != CHUNK_SIZE * 2:
                    print(f"[WARN] 音频数据长度不匹配: {len(audio_data)} vs {CHUNK_SIZE*2}")
                    # 调整数据长度
                    if len(audio_data) > CHUNK_SIZE * 2:
                        # 截断过长的数据
                        audio_data = audio_data[:CHUNK_SIZE * 2]
                    else:
                        # 填充不足的数据
                        audio_data = audio_data + b'\x00' * (CHUNK_SIZE * 2 - len(audio_data))
                
                # 播放音频数据
                try:
                    # 对于第一个音频包，确保优先播放
                    if is_first_audio_packet:
                        print("[AUDIO] 优先处理第一个音频包")
                        is_first_audio_packet = False
                        # 直接播放第一个包
                        stream.write(audio_data)
                    else:
                        await asyncio.to_thread(stream.write, audio_data)
                    
                    # if play_count % 10 == 0:
                    #     print(f"[AUDIO] 已播放 {play_count} 个音频块，队列剩余: {AUDIO_QUEUE.qsize()}")  # 注意：在高负载时，获取队列大小可能影响性能
                except Exception as write_e:
                    print(f"[ERROR] 播放音频数据时出错: {write_e}")
                    # 尝试重新启动流
                    try:
                        stream.stop_stream()
                        stream.start_stream()
                        print("[RESTART] 已重新启动音频流")
                    except:
                        pass
                
                # 标记任务完成
                AUDIO_QUEUE.task_done()
                
            except asyncio.TimeoutError:
                # 超时但不退出，继续等待新数据
                continue
            except Exception as play_e:
                print(f"[ERROR] 音频播放过程中出错: {play_e}")
                # 确保队列任务完成
                try:
                    AUDIO_QUEUE.task_done()
                except:
                    pass
                # 短暂暂停后继续
                await asyncio.sleep(0.01)
                continue
            
            # 如果连续收到空数据，尝试重新初始化音频设备
            if consecutive_empty_data >= 5:
                print(f"[WARN] 连续收到 {consecutive_empty_data} 个空数据块，重新初始化音频设备...")
                try:
                    stream.stop_stream()
                    stream.close()
                    stream = p.open(
                        format=current_format,
                        channels=current_channels,
                        rate=current_rate,
                        output=True,
                        frames_per_buffer=CHUNK_SIZE
                    )
                    stream.start_stream()
                    consecutive_empty_data = 0
                    print("[OK] 音频设备已重新初始化")
                except Exception as reset_e:
                    print(f"[ERROR] 重新初始化音频设备失败: {reset_e}")
                    consecutive_empty_data = 0
    except Exception as e:
        print(f"[ERROR] 音频播放任务出错: {e}")
        import traceback
        traceback.print_exc()
    finally:
        stream_active = False
        if stream is not None:
            try:
                stream.stop_stream()
                stream.close()
                print("[OK] 音频流已关闭")
            except Exception as close_e:
                print(f"[ERROR] 关闭音频流时出错: {close_e}")
        if p is not None:
            try:
                p.terminate()
                print("[OK] PyAudio已终止")
            except Exception as term_e:
                print(f"[ERROR] 终止PyAudio时出错: {term_e}")
        
        # 清空队列中的剩余数据
        cleared_count = 0
        # 清理音频队列的异步方式 - 分批处理避免阻塞
        for _ in range(50):  # 限制清理数量，避免长时间阻塞
            try:
                await asyncio.wait_for(AUDIO_QUEUE.get(), timeout=0.01)
                AUDIO_QUEUE.task_done()
                cleared_count += 1
            except (asyncio.TimeoutError, asyncio.QueueEmpty):
                break
            except:
                break
        print(f"[OK] 已清空队列中的 {cleared_count} 个音频块")

# 单独的消息监听任务：持续接收服务器消息
async def listen_messages(websocket):
    # 确保所有必要的导入都在函数开始处
    import time
    import json
    audio_packet_count = 0
    last_audio_time = time.time()
    consecutive_audio_blocks = 0
    
    try:
        while True:
            msg = await websocket.recv()  # 阻塞等待消息
            current_time = time.time()
            
            if isinstance(msg, str):
                print(f"\n📄 收到服务器消息: {msg}")
                # 解析JSON（如果需要）
                try:
                    msg_json = json.loads(msg)
                    # print(f"🔍 解析后的消息类型: {msg_json.get('type', 'unknown')}")
                    
                    # 检查是否为ping/pong消息
                    if msg_json.get('type') == 'ping':
                        # 发送pong响应
                        await websocket.send(json.dumps({"type": "pong"}))
                        print("🏓 已响应ping消息")
                    elif msg_json.get('type') == 'pong':
                        print("🏓 收到pong消息，连接正常")
                    
                    if 'error' in msg_json:
                        print(f"❌ 错误消息: {msg_json['error']}")
                    
                    # 显示会话信息
                    # if 'session_id' in msg_json:
                    #     print(f"✅ 会话ID: {msg_json['session_id']}")
                except json.JSONDecodeError:
                    print("📄 消息为普通字符串")
                
                # 重置音频块计数
                consecutive_audio_blocks = 0
            # 收到二进制音频数据
            elif isinstance(msg, bytes):
                audio_packet_count += 1
                last_audio_time = current_time
                consecutive_audio_blocks += 1
                
                # print(f"🎵 收到音频数据包 #{audio_packet_count}，长度: {len(msg)} 字节")
                
                # 处理音频数据
                try:
                    # 解码Opus音频
                    pcm_data = await asyncio.to_thread(process_opus_audio, msg)
                    # print(f"📊 处理后音频数据长度: {len(pcm_data)} 字节")
                except Exception as process_error:
                    print(f"❌ 处理音频数据时出错: {process_error}")
                    # 即使处理出错，也创建一个简单的提示音
                    import random
                    pcm_data = bytes(random.randint(-100, 100) for _ in range(CHUNK_SIZE * 2))
                
                # 即使pcm_data为空，也放入队列一个提示音
                if not pcm_data or len(pcm_data) == 0:
                    print("⚠️  处理后音频数据为空，创建提示音")
                    import random
                    pcm_data = bytes(random.randint(-100, 100) for _ in range(CHUNK_SIZE * 2))
                
                # 检查队列状态，防止积压
                # 注意：在高负载时，获取队列大小可能导致性能问题
                queue_size = AUDIO_QUEUE.qsize()
                
                # 对于第一个音频包，优先处理
                global is_first_audio_packet
                if is_first_audio_packet:
                    # 第一个音频包特殊处理，确保能被立即播放
                    print(f"[AUDIO] 处理第一个音频包，队列当前大小: {queue_size}")
                    
                    # 关键改进：清空队列并直接播放第一个包，不经过队列
                    # 这样可以避免队列引入的任何延迟
                    if queue_size > 0:
                        cleared_count = 0
                        # 清理音频队列
                        for _ in range(50):  # 限制清理数量
                            try:
                                await asyncio.wait_for(AUDIO_QUEUE.get(), timeout=0.01)
                                AUDIO_QUEUE.task_done()
                                cleared_count += 1
                            except (asyncio.TimeoutError, asyncio.QueueEmpty, queue.Empty):
                                break
                        print(f"[AUDIO] 已清空 {cleared_count} 个旧数据，确保新音频数据优先播放")
                    
                    # 简化处理：第一个包也通过队列处理，避免代码重复
                    # 这样可以保持统一的音频处理流程
                else:
                    # 队列积压处理机制
                    if queue_size >= MAX_QUEUE_SIZE:
                        # 队列已满，清空整个队列以避免长时间阻塞
                        print(f"[WARN] 音频队列已满({queue_size}/{MAX_QUEUE_SIZE})，清空队列以恢复正常播放")
                        cleared_count = 0
                        # 清理音频队列
                        for _ in range(50):  # 限制清理数量
                            try:
                                await asyncio.wait_for(AUDIO_QUEUE.get(), timeout=0.01)
                                AUDIO_QUEUE.task_done()
                                cleared_count += 1
                            except (asyncio.TimeoutError, asyncio.QueueEmpty, queue.Empty):
                                break
                        print(f"[OK] 已清空 {cleared_count} 个旧音频块")
                    elif queue_size > MAX_QUEUE_SIZE * 0.7:  # 队列超过70%容量时清理
                        # 队列积压，丢弃一部分数据
                        discard_count = max(1, queue_size // 3)  # 丢弃约1/3的数据
                        print(f"[WARN] 音频队列积压({queue_size}/{MAX_QUEUE_SIZE})，丢弃{discard_count}个旧数据块")
                        for _ in range(discard_count):
                            try:
                                # 使用超时为0.01秒的异步获取
                                await asyncio.wait_for(AUDIO_QUEUE.get(), timeout=0.01)
                                AUDIO_QUEUE.task_done()
                            except (asyncio.TimeoutError, asyncio.QueueEmpty, queue.Empty):
                                break
                
                # 将处理后的音频数据放入队列
                try:
                    # 对于第一个音频包，使用更高的优先级
                    if is_first_audio_packet:
                        # print("[AUDIO] 第一个音频包，优先处理")
                        # 如果队列为空，立即放入；如果队列不为空，先清理一个位置
                        if AUDIO_QUEUE.qsize() > 0:
                                try:
                                    # 使用超时为0.01秒的异步获取
                                    await asyncio.wait_for(AUDIO_QUEUE.get(), timeout=0.01)
                                    AUDIO_QUEUE.task_done()
                                    print("[AUDIO] 已清理队列中的旧数据，为第一个音频包腾出空间")
                                except (asyncio.TimeoutError, asyncio.QueueEmpty):
                                    pass
                                except:
                                    pass
                    
                    # 非阻塞方式添加数据，避免队列满时阻塞
                    try:
                        # 使用try/except捕获队列满的情况
                        await asyncio.wait_for(AUDIO_QUEUE.put(pcm_data), timeout=0.1)
                    except asyncio.QueueFull:
                        print("[WARN] 音频队列已满，丢弃数据")
                    # print(f"[OK] 音频数据已加入队列，当前队列大小: {AUDIO_QUEUE.qsize()}")
                except queue.Full:
                    print("[ERROR] 音频队列已满，丢弃部分数据")
                    # 尝试清理一个位置并添加新数据
                    try:
                        # 尝试清理一个位置
                        await asyncio.wait_for(AUDIO_QUEUE.get(), timeout=0.01)
                        AUDIO_QUEUE.task_done()
                        # 尝试添加新数据
                        await asyncio.wait_for(AUDIO_QUEUE.put(pcm_data), timeout=0.1)
                        print("[OK] 已清理一个位置并添加新数据")
                    except (asyncio.TimeoutError, asyncio.QueueEmpty):
                        pass
                    except Exception as put_e:
                        print(f"[ERROR] 将音频数据加入队列失败: {put_e}")
                
                # 定期打印队列状态
                # if consecutive_audio_blocks % 5 == 0:
                #     print(f"📊 音频流统计 - 已接收: {audio_packet_count} 块, 队列大小: {AUDIO_QUEUE.qsize()}")
            
            # 检查是否有一段时间没有收到音频数据
            if current_time - last_audio_time > 5 and consecutive_audio_blocks > 0:
                print(f"⏸️  音频流中断，已{current_time - last_audio_time:.1f}秒未收到音频数据")
                consecutive_audio_blocks = 0
                
    except asyncio.TimeoutError:
        # 超时但不退出，继续等待新数据
        print("⏱️  接收消息超时，继续等待...")
    except websockets.exceptions.ConnectionClosed as e:
        print(f"🔌 连接已关闭，停止监听 - 原因: {e}")
    except Exception as e:
        print(f"❌ 监听出错: {e}")
        import traceback
        traceback.print_exc()
    finally:
        print(f"📋 监听任务结束 - 共接收 {audio_packet_count} 个音频包")

# 辅助函数：PCM音频处理类
class AudioProcessor:
    def __init__(self, buffer_size=960):
        self.buffer_size = buffer_size
        self.buffer = np.array([], dtype=np.float32)
        self.sample_rate = 16000
    
    def reset_buffer(self):
        self.buffer = np.array([], dtype=np.float32)
    
    def process_audio(self, input_data):
        # 将输入数据转换为float32数组
        # 假设input_data是int16格式的PCM数据
        input_array = np.frombuffer(input_data, dtype=np.int16).astype(np.float32) / 32767.0
        
        # 将新数据添加到缓冲区
        self.buffer = np.append(self.buffer, input_array)
        
        chunks = []
        # 当缓冲区达到指定大小时处理数据
        while len(self.buffer) >= self.buffer_size:
            # 提取数据
            chunk = self.buffer[:self.buffer_size]
            self.buffer = self.buffer[self.buffer_size:]
            
            # 转回int16格式
            pcm_data = (chunk * 32767).astype(np.int16)
            chunks.append(pcm_data.tobytes())
        
        return chunks
    
    def process_remaining(self):
        if len(self.buffer) > 0:
            # 转回int16格式
            pcm_data = (self.buffer * 32767).astype(np.int16)
            self.buffer = np.array([], dtype=np.float32)
            return [pcm_data.tobytes()]
        return []

# 辅助函数：将PCM数据转换为Opus格式
def pcm_to_opus(pcm_data):
    """将PCM音频数据转换为Opus格式"""
    try:
        if not OPUS_AVAILABLE:
            print("[WARN] Opus不可用，无法编码音频数据")
            return None
        
        # 创建编码器：16kHz, 单声道, VOIP模式
        encoder = opuslib.Encoder(16000, 1, 'voip')
        
        try:
            # 确保PCM数据是Int16格式
            pcm_array = np.frombuffer(pcm_data, dtype=np.int16)
            
            # 编码PCM数据，每帧960个采样点
            opus_data = encoder.encode(pcm_array.tobytes(), 960)  # 60ms at 16kHz
            return opus_data
            
        except Exception as e:
            print(f"Opus编码错误: {e}, 数据长度: {len(pcm_data)}")
            return None
            
    except Exception as e:
        print(f"Opus初始化错误: {e}")
        return None

# 发送录音数据到websocket
async def send_recording_to_websocket(websocket, recorded_data):
    """将录制的音频数据发送到websocket，按照proxy.py的逻辑"""
    print("[WEBSOCKET] 开始发送录音数据...")
    
    try:
        # 发送开始消息
        start_msg = {
            "type": "audio",
            "state": "start"
        }
        await websocket.send(json.dumps(start_msg))
        print("[WEBSOCKET] 已发送音频开始消息")
        
        # 创建音频处理器
        audio_processor = AudioProcessor(buffer_size=960)
        
        # 处理所有录音数据
        total_chunks = 0
        for audio_data in recorded_data:
            # 使用AudioProcessor处理音频数据
            chunks = audio_processor.process_audio(audio_data)
            for chunk in chunks:
                # 转换为Opus格式
                opus_data = pcm_to_opus(chunk)
                if opus_data:
                    # 直接以二进制形式发送Opus数据
                    await websocket.send(opus_data)
                    total_chunks += 1
                    
                    # 每10个块显示一次进度
                    if total_chunks % 10 == 0:
                        print(f"[WEBSOCKET] 已发送 {total_chunks} 个Opus音频块")
        
        # 处理剩余数据
        remaining_chunks = audio_processor.process_remaining()
        for chunk in remaining_chunks:
            opus_data = pcm_to_opus(chunk)
            if opus_data:
                await websocket.send(opus_data)
                total_chunks += 1
        
        # 发送结束消息
        stop_msg = {
            "type": "audio",
            "state": "stop"
        }
        await websocket.send(json.dumps(stop_msg))
        
        print(f"[WEBSOCKET] 录音数据发送完成，共发送 {total_chunks} 个Opus音频块")
    except Exception as e:
        print(f"[ERROR] 发送录音数据时出错: {e}")

# 使用传入数据播放录制的音频
def play_recorded_audio_with_data(recorded_data):
    """播放录制的音频数据"""
    print("[AUDIO] 开始播放录制的音频...")
    
    # 检查是否有录制的数据
    if not recorded_data:
        print("[AUDIO] 没有录制的音频数据")
        return
    
    print(f"[AUDIO] 准备播放 {len(recorded_data)} 个音频块")
    
    try:
        # 创建独立的PyAudio实例用于播放
        p = pyaudio.PyAudio()
        
        # 打开音频输出流
        stream = p.open(
            format=DEFAULT_FORMAT,
            channels=DEFAULT_CHANNELS,
            rate=DEFAULT_RATE,
            output=True,
            frames_per_buffer=CHUNK_SIZE
        )
        
        # 播放所有录制的音频数据
        for i, data in enumerate(recorded_data):
            stream.write(data)
            # 每10个块显示一次进度
            if (i + 1) % 10 == 0:
                print(f"[AUDIO] 已播放 {i + 1}/{len(recorded_data)} 个音频块")
        
        print("[AUDIO] 录制音频播放完成")
        
        # 关闭流和PyAudio实例
        stream.stop_stream()
        stream.close()
        p.terminate()
        
    except Exception as e:
        print(f"[ERROR] 播放录制的音频时出错: {e}")
    finally:
        # 确保清空队列，避免数据残留
        try:
            while not RECORDING_QUEUE.empty():
                RECORDING_QUEUE.get_nowait()
                RECORDING_QUEUE.task_done()
            print("[AUDIO] 录音队列已清空")
        except:
            pass
    


# 单独的消息发送任务：随时可向服务器发送消息
async def send_messages(websocket):
    try:
        # 示例：循环输入消息发送（也可以是程序自动触发）
        while True:
            # 等待用户输入（模拟业务触发）
            user_input = await asyncio.to_thread(input, "\n请输入要发送的消息（输入q退出，r开始/停止录音）: ")
            
            # 检查是否为退出命令
            if user_input.lower() == 'q':
                print("退出发送任务")
                # 设置一个全局标志表示用户主动退出
                global user_initiated_exit
                user_initiated_exit = True
                break
            
            # 检查是否为录音控制命令
            elif user_input.lower() == 'r':
                global is_recording
                is_recording = not is_recording
                if is_recording:
                    print("🎤 开始录音... (再次按'r'停止)")
                    # 清空之前的录音队列
                    await clear_audio_queue(RECORDING_QUEUE)
                else:
                    print("🎤 停止录音...")
                    # 给录音任务一点时间完成最后的数据写入
                    await asyncio.sleep(0.1)
                    
                    # 收集录音数据（用于播放和发送）
                    recorded_data = []
                    while not RECORDING_QUEUE.empty():
                        try:
                            data = RECORDING_QUEUE.get_nowait()
                            recorded_data.append(data)
                            RECORDING_QUEUE.task_done()
                        except:
                            break
                    
                    # 播放录制的音频
                    if recorded_data:
                        print(f"🔊 正在播放录制的音频... (共{len(recorded_data)}个数据块)")
                        await asyncio.to_thread(lambda: play_recorded_audio_with_data(recorded_data))
                        print("✅ 音频播放完成")
                        
                        # 发送录音数据到websocket
                        print("📤 正在发送录音数据到websocket...")
                        await send_recording_to_websocket(websocket, recorded_data)
                        print("✅ 录音数据发送完成")
                    else:
                        print("⚠️  未录制到任何音频数据")
                continue
            
            # 发送普通消息
            # 发送消息（可以是字符串或JSON）
            dataJson = { 
                "type": "listen",
                "state": "detect",
                "text": user_input,
                "source": "text" 
            }
            
            await websocket.send(json.dumps(dataJson))
            print(f"已发送: {user_input}")
    except Exception as e:
        print(f"发送出错: {e}")

# 主连接逻辑：同时启动监听和发送任务
import time

# 全局标志：用户是否主动退出
user_initiated_exit = False

# 主连接逻辑：同时启动监听和发送任务
async def connect_websocket() -> str:    
    websocket_url = "wss://api.tenclass.net/xiaozhi/v1/"
    token = "test-token"
    session_id = ''
    MAX_RECONNECT_ATTEMPTS = 5  # 增加最大重连次数
    reconnect_attempts = 0
    connection_established = False
    
    # 重置用户退出标志
    global user_initiated_exit
    user_initiated_exit = False
    
    headers = {
        "Authorization": f"Bearer {token}",
        "Protocol-Version": "1",
        "Device-Id": "94:a9:90:1a:e2:08",
        "Client-Id": "device-uuid-1234567"
    }
    
    # 清空音频队列的函数 - 异步版本
    async def clear_audio_queue(queue_to_clear=None):
        # 默认清理AUDIO_QUEUE，也可以指定其他队列
        queue = queue_to_clear if queue_to_clear is not None else AUDIO_QUEUE
        # 注意：asyncio.Queue没有empty()方法，但我们可以尝试检查大小
        cleared_count = 0
        max_clear_per_cycle = 50
        
        # 尝试清理队列，限制每次清理的项目数
        for _ in range(max_clear_per_cycle):
            try:
                # 使用超时为0的等待，模拟非阻塞操作
                await asyncio.wait_for(queue.get(), timeout=0.01)
                queue.task_done()
                cleared_count += 1
            except (asyncio.TimeoutError, asyncio.QueueEmpty):
                break
        
        if cleared_count > 0:
            print(f"[AUDIO] 已清空队列中的 {cleared_count} 个数据块")
        else:
            print("音频队列为空，无需清理")
            
        # 如果队列中仍有项目，安排下次清理
        if cleared_count >= max_clear_per_cycle:
            print("[DEBUG] 队列项目过多，继续清理...")
            asyncio.create_task(clear_audio_queue(queue))
            
        # 重置第一个包标记
        global is_first_audio_packet
        is_first_audio_packet = True

    # 心跳任务：定期发送ping保持连接活跃
    async def heartbeat(websocket):
        try:
            while True:
                await asyncio.sleep(20)  # 20秒发送一次心跳
                try:
                    # 使用send_ping方法而非自定义ping消息
                    await websocket.ping()
                    # print("[DEBUG] 发送心跳包")
                except Exception as ping_e:
                    print(f"[WARN] 发送心跳包失败: {ping_e}")
                    break
        except asyncio.CancelledError:
            pass

    while True:
        try:
            # 优化WebSocket连接参数
            async with websockets.connect(
                websocket_url, 
                extra_headers=headers,
                ping_interval=20,      # 减少ping间隔以保持连接活跃
                ping_timeout=15,       # 增加超时时间
                close_timeout=5,       # 添加关闭超时
                open_timeout=20,       # 添加连接超时
                max_size=2**20,        # 增加最大消息大小到1MB
                compression=None       # 禁用压缩以提高兼容性
            ) as websocket:
                print("连接成功！")
                connection_established = True
                reconnect_attempts = 0  # 重置重连计数
                
                # 连接成功后清空音频队列
                await clear_audio_queue()

                # 发送hello消息
                hello_msg = {
                    "type": "hello",
                    "version": 1,
                    "features": {"mcp": False},
                    "transport": "websocket",
                    "audio_params": {
                        "format": "opus",        # 服务器间传输使用Opus编码
                        "sample_rate": 16000,    # 采样率16kHz
                        "channels": 1,           # 单声道
                        "frame_duration": 60     # 帧持续时间60ms，对应960样本
                    }
                }
                await websocket.send(json.dumps(hello_msg))
                print("已发送hello消息")

                # 等待hello应答并获取session_id
                try:
                    response = await asyncio.wait_for(websocket.recv(), timeout=10.0)
                    if isinstance(response, str):
                        try:
                            response_json = json.loads(response)
                            print(f"收到hello应答: {response_json}")
                            session_id = response_json.get('session_id')
                        except json.JSONDecodeError:
                            print("[ERROR] hello应答无法解析为JSON")
                    else:
                        print("[ERROR] hello应答不是字符串格式")
                except asyncio.TimeoutError:
                    print("[WARN] 超时未收到hello应答，继续运行")

                # 启动心跳任务
                heartbeat_task = asyncio.create_task(heartbeat(websocket), name="heartbeat")

                # 关键：创建四个并发任务（监听、发送、音频播放和录音）
                tasks = [heartbeat_task]
                
                # 1. 启动音频播放任务（先启动确保音频设备准备好）
                audio_task = asyncio.create_task(play_audio(), name="audio_playback")
                tasks.append(audio_task)
                
                # 2. 启动录音任务（后台运行，但默认不录音）
                record_task = asyncio.create_task(record_audio(), name="audio_recording")
                tasks.append(record_task)
                
                # 3. 启动监听任务（后台运行）
                listen_task = asyncio.create_task(listen_messages(websocket), name="message_listener")
                tasks.append(listen_task)
                
                # 4. 启动发送任务（前台运行，可交互）
                send_task = asyncio.create_task(send_messages(websocket), name="message_sender")
                tasks.append(send_task)

                # 等待发送任务结束（用户输入q时）或任一任务出错
                done, pending = await asyncio.wait(
                    tasks,
                    return_when=asyncio.FIRST_COMPLETED,
                    timeout=None
                )
                
                # 检查完成的任务是否有异常
                error_occurred = False
                for task in done:
                    if task.get_name() != "message_sender" and task.exception():
                        error_type = type(task.exception()).__name__
                        print(f"[ERROR] 任务 {task.get_name()} 出现异常: {task.exception()}")
                        print(f"[ERROR] 异常类型: {error_type}")
                        # 特别处理WebSocket连接错误
                        if "ConnectionClosed" in error_type:
                            print("[WARN] WebSocket连接已关闭，将尝试重连")
                        error_occurred = True
                
                # 取消所有剩余的任务
                for task in pending:
                    task.cancel()
                    try:
                        await task
                    except asyncio.CancelledError:
                        print(f"任务 {task.get_name()} 已取消")
                
                # 只有当用户输入q主动退出时才不重连
                # 对于TTS停止消息导致的连接关闭，我们应该保持重连
                if not error_occurred:
                    print("[INFO] 检测到可能是任务完成导致的关闭，准备重新连接以保持对话状态...")
                    # 不break，继续重连循环
                    # 这样即使收到TTS停止消息后连接关闭，也会自动重新连接

        except websockets.exceptions.ConnectionClosed as e:
            # 专门处理WebSocket关闭错误
            print(f"[ERROR] WebSocket连接已关闭: {e}")
            # 特别处理1005错误码
            if "1005" in str(e):
                print("[WARN] 检测到1005错误（内部错误），将尝试重连")
            reconnect_attempts += 1
        except websockets.exceptions.ConnectionClosedError as e:
            # 处理连接关闭错误
            print(f"[ERROR] WebSocket连接错误: {e}")
            reconnect_attempts += 1
        except websockets.exceptions.ConnectionClosedOK as e:
            # 正常关闭连接
            print(f"[INFO] WebSocket连接已正常关闭: {e}")
            # 修改：即使是正常关闭也尝试重连，以保持对话状态
            print("[INFO] 准备重新连接以保持对话状态...")
            reconnect_attempts += 1
        except Exception as e:
            # 记录详细的错误信息
            error_type = type(e).__name__
            print(f"[ERROR] 连接出错: {e}")
            print(f"[ERROR] 错误类型: {error_type}")
            reconnect_attempts += 1
        finally:
            # 确保音频播放任务停止
            await clear_audio_queue()
            connection_established = False
        
        # 检查用户是否主动退出
        if user_initiated_exit:
            print("[INFO] 用户主动退出，停止重连")
            break
            
        # 检查是否达到最大重连次数
        if reconnect_attempts > MAX_RECONNECT_ATTEMPTS:
            print(f"[ERROR] 已达到最大重连次数({MAX_RECONNECT_ATTEMPTS})，停止重连")
            break
        
        # 智能退避重连：结合指数退避和随机延迟避免雪崩效应
        base_delay = min(15, 2 ** (reconnect_attempts - 1))  # 最大15秒
        jitter = random.uniform(0.5, 1.5)  # 添加随机抖动
        reconnect_delay = base_delay * jitter
        print(f"[INFO] {reconnect_delay:.1f}秒后尝试第{reconnect_attempts}次重连...")
        await asyncio.sleep(reconnect_delay)
    
    return session_id

# 测试音频播放功能
def test_audio_playback():
    """测试音频播放功能是否正常工作"""
    print("\n[TEST] 开始音频播放测试...")
    
    try:
        # 创建PyAudio实例
        p = pyaudio.PyAudio()
        
        # 打开音频流
        stream = p.open(
            format=DEFAULT_FORMAT,
            channels=DEFAULT_CHANNELS,
            rate=DEFAULT_RATE,
            output=True,
            frames_per_buffer=CHUNK_SIZE
        )
        
        print(f"[OK] 音频测试流已打开 - {DEFAULT_RATE}Hz, {DEFAULT_CHANNELS}声道")
        
        # 生成一个简单的正弦波
        import numpy as np
        duration = 0.5  # 0.5秒的测试音
        frequency = 440  # A4音
        samples = int(DEFAULT_RATE * duration)
        t = np.linspace(0, duration, samples, endpoint=False)
        
        # 创建正弦波并转换为16位PCM
        sine_wave = np.sin(2 * np.pi * frequency * t)
        # 调整音量到10%
        sine_wave = sine_wave * 0.1
        # 转换为16位整数
        pcm_data = np.int16(sine_wave * 32767)
        
        # 播放测试音
        print("[AUDIO] 播放测试音频 (440Hz 正弦波)")
        stream.write(pcm_data.tobytes())
        
        # 关闭流
        stream.stop_stream()
        stream.close()
        p.terminate()
        
        print("[OK] 音频测试成功完成！")
        return True
    except Exception as e:
        print(f"[ERROR] 音频测试失败: {e}")
        import traceback
        traceback.print_exc()
        return False

# 执行主函数
if __name__ == "__main__":
    print("=== XiaoZhi Python 客户端启动 ===")
    print("[INIT] 初始化音频系统...")
    
    # 先进行音频播放测试
    audio_test_passed = test_audio_playback()
    
    print("\n[INFO] 系统信息:")
    print(f"  - Opus解码可用: {'YES' if OPUS_AVAILABLE else 'NO'}")
    print(f"  - 音频设备测试: {'PASSED' if audio_test_passed else 'FAILED'}")
    
    print("\n[NOTE] 提示:")
    print("1. 确保已安装所需库: pip install pyaudio opuslib==3.0.1 numpy websockets")
    print("2. 音频配置: 16kHz采样率, 单声道, 16位PCM格式")
    print("3. 即使Opus解码不可用，程序也会播放提示音表示收到音频数据")
    print("4. 按q退出程序")
    print("5. 按r开始/停止录音")
    print("======================================\n")
    
    try:
        session_id = asyncio.run(connect_websocket())
        print(f"[DONE] 所有操作完成, session_id: {session_id}")
    except KeyboardInterrupt:
        print("\n[EXIT] 程序已强制退出")
    except Exception as e:
        print(f"\n[ERROR] 程序运行出错: {e}")
        import traceback
        traceback.print_exc()
    finally:
        print("\n[END] 程序已结束")